#include "net/base/client_connection.h"
#include <set>
#include <arpa/inet.h>
#include <errno.h>
#include <stdio.h>
#include <poll.h>
#include <string.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <unistd.h>
#include "base/callback.h"
#include "base/logging.h"
#include "base/scoped_ptr.h"
#include "thread/executor.h"
 
using std::set;

/*
ClientConnection::ClientConnection(const char* ip, int port)
    : ip_(ip), port_(port), status_(UNKNOWN), sockfd_(-1) {
}

ClientConnection::~ClientConnection() {
  if (sockfd_ != -1) {
    close(sockfd_);
  }
}

void ClientConnection::Connect() {
  sockfd_ = socket(AF_INET, SOCK_STREAM, 0);
  int value = 1;
  setsockopt(sockfd_, SOL_SOCKET, SO_KEEPALIVE, &value, sizeof(value));
  if (sockfd_ < 0) {
    perror("Failed to create socket");
    LOG(FATAL) << "Failed to create socket";
  }
  sockaddr_in addr;
  memset(&addr, 0, sizeof(sockaddr_in));
  addr.sin_family = AF_INET;
  addr.sin_port = htons(port_);
   
  if((addr.sin_addr.s_addr = inet_addr(ip_.c_str())) == (unsigned long)INADDR_NONE) {
    LOG(FATAL) << "Invalid ip " << ip_;
  }
  int conn_result = connect(sockfd_, (sockaddr*) &addr, sizeof(addr));
  if (conn_result < 0) {
    perror("Client-connect() error");
    LOG(FATAL) << "Failed to connect to " << ip_;
  }
  status_ = CONNECTED;
}

bool ClientConnection::Send(const char* buffer, int size) {
  if (status_ != CONNECTED) {
    LOG(ERROR) << "Not connected";
    return false;
  }
  const char* pt = buffer;
  int size_left = size;
  while (size_left > 0) {
    // ssize_t result = send(sockfd_, pt, size_left, MSG_DONTWAIT);
    ssize_t result = send(sockfd_, pt, size_left, MSG_NOSIGNAL);
    if (result == -1) {
      perror("Error sending");
      LOG(ERROR) << "Could not send";
      Close();
      return false;
    } else {
      pt += result;
      size_left -= result;
    }
  }
  return true;
}

int ClientConnection::Recv(char* buffer, int size) {
  if (status_ != CONNECTED) {
    LOG(ERROR) << "Not connected";
    return false;
  }
  int actual_size = recv(sockfd_, buffer, size, 0);
  if (actual_size == -1) {
    perror("Failed to recv");
    LOG(ERROR) << "Failed to recv ";
  } else if (actual_size == 0) {
    perror("Failed to recv");
    LOG(ERROR) << "Failed to recv ";
    Close();
  }
  return actual_size;
}

void ClientConnection::Close() {
  close(sockfd_);
  sockfd_ = -1;
  status_ = DISCONNECTED;
}

int ClientConnection::sockfd() {
  return sockfd_;
}*/

ClientPollRunner::ClientPollRunner() : exit_(false) {
}

void ClientPollRunner::AddConnection(AsyncClientConnection* conn) {
  WriteLock lock(&conn_mutex_);
  conn_.push_back(conn);
}

void ClientPollRunner::RemoveConnection(AsyncClientConnection* conn) {
  WriteLock lock(&conn_mutex_);
  for (int i = 0; i < conn_.size(); ++i) {
    if (conn_[i] ==conn) {
      conn_.erase(conn_.begin() + i);
      break;
    }
  }
}

void ClientPollRunner::Loop() {
  while (!exit_) {
    pollfd* polls;
    int polls_size = 0;
    {
      ReadLock rl(&conn_mutex_);
      polls_size = conn_.size();
      polls = new pollfd[polls_size];
      for (int i = 0; i < conn_.size(); ++i) {
        polls[i].fd = conn_[i]->sockfd();
        polls[i].events = POLLIN | POLLPRI | POLLRDHUP;
        if (conn_[i]->send_size() > 0) {
          polls[i].events |= POLLOUT;
        }
      }
    }
    int r = poll(polls, polls_size, 2000);
    if (r == 0) {
      delete[] polls;
      // Timeout.
      continue;
    }
    if (r == -1) {
      LOG(ERROR) << "Error poll";
      // TODO: errorno.
      delete[] polls;
      continue;
    }
    for (int i = 0; i < polls_size; ++i) {
      if (polls[i].revents & POLLIN) {
        // TODO: MSGOOB. Handle error.
        int size = 1024 * 1024;
        char* buffer = new char[size];
        size = recv(polls[i].fd, buffer, size, MSG_DONTWAIT);
        if (size < 0) {
          // TODO:errorno
          LOG(ERROR) << "Error receiving data";
          perror("erro");
          delete[] buffer;
        } else if (size == 0) {
          conn_[i]->status_ = DISCONNECTED;
          continue;
        } else {
          conn_[i]->HandleRecv(buffer, size);
        }
      }
      if (polls[i].revents & POLLRDHUP ||
          polls[i].revents & POLLHUP) {
        conn_[i]->status_ = DISCONNECTED;
        continue;
      }
      if (polls[i].revents & POLLNVAL) {
        LOG(ERROR) << "Not open";
      }
      if (polls[i].revents & POLLOUT) {
        conn_[i]->HandleWrite();
      }
    }
    CleanDisconnected();
    delete[] polls;
  }
}

void ClientPollRunner::CleanDisconnected() {
  WriteLock wl(&conn_mutex_);
  for (int i = conn_.size() - 1; i >= 0; --i) {
    if (conn_[i]->status_ == DISCONNECTED) {
      conn_[i]->HandleDisconnect();
      conn_.erase(conn_.begin() + i);
    }
  }
}

AsyncClientConnection::AsyncClientConnection(const char* ip, int port)
    : status_(UNKNOWN), ip_(ip), port_(port), sockfd_(-1) {
}

AsyncClientConnection::~AsyncClientConnection() {
  if (status_ != DISCONNECTED) {
    Close();
  }
}

void AsyncClientConnection::Connect() {
  sockfd_ = socket(AF_INET, SOCK_STREAM, 0);
  int value = 1;
  setsockopt(sockfd_, SOL_SOCKET, SO_KEEPALIVE, &value, sizeof(value));
  if (sockfd_ < 0) {
    perror("Failed to create socket");
    LOG(FATAL) << "Failed to create socket";
  }
  sockaddr_in addr;
  memset(&addr, 0, sizeof(sockaddr_in));
  addr.sin_family = AF_INET;
  addr.sin_port = htons(port_);
   
  if((addr.sin_addr.s_addr = inet_addr(ip_.c_str())) == (unsigned long)INADDR_NONE) {
    LOG(FATAL) << "Invalid ip " << ip_;
  }
  int conn_result = connect(sockfd_, (sockaddr*) &addr, sizeof(addr));
  if (conn_result < 0) {
    perror("Client-connect() error");
    LOG(FATAL) << "Failed to connect to " << ip_;
  }
  status_ = CONNECTED;
  poll_runner_->AddConnection(this);
}

int AsyncClientConnection::send_size() {
  ReadLock lock(&byte_stream_mutex_);
  return byte_stream_.size();
}

bool AsyncClientConnection::Send(const char* buffer, int size) {
  if (status_ != CONNECTED) {
    LOG(ERROR) << "Not connected";
    return false;
  }
  WriteLock lock(&byte_stream_mutex_);
  byte_stream_.Push(buffer, size);
  return true;
}

void AsyncClientConnection::Close() {
  poll_runner_->RemoveConnection(this);
  close(sockfd_);
  sockfd_ = -1;
  status_ = DISCONNECTED;
}

int AsyncClientConnection::sockfd() {
  return sockfd_;
}

void AsyncClientConnection::SetPollRunner(ClientPollRunner* poll_runner) {
  poll_runner_ = poll_runner;
}

void AsyncClientConnection::HandleWrite() {
  WriteLock lock(&byte_stream_mutex_);
  if (byte_stream_.size() == 0) return;
  int buff_size = std::min(2 << 16, byte_stream_.size());
  char* buffer = new char[buff_size];
  byte_stream_.Peek(buffer, buff_size);
  ssize_t result = send(sockfd_, buffer, buff_size, MSG_NOSIGNAL);
  if (result == -1) {
    perror("Error sending");
  } else {
    byte_stream_.Pop(result);
  }
}
